﻿#include "net.h"
#include "queue.h"

// 滑动窗口的结构体, 队首front即窗口左侧先发的数据, 队尾rear即窗口右侧后发的数据
struct Struct_Queue *receive_window;
// 滑动窗口互斥量, 保证只有一个线程在访问发送窗口中的数据
pthread_mutex_t lock;
// 记录最后被滑动走的序列号, 比如当前窗口为[11, 45], 则该变量记录为10
int last_slide_sequence = 0;
// 记录开始标志, 用于发送第一帧
int begin_flag = 0;
// 结束包的指针, 用于接收最后一帧
struct Struct_Queue_Node *finish_ptr = NULL;
// 记录结束标志, 用于发送最后一帧
int finish_flag = 0;
// 总共接收的字节数
int total_receive_byte = 0;
// 有效接收的字节数
int valid_receive_byte = 0;

/**
 * @brief 用于接收DATA包的线程函数
 * 
 * @param arg 参量表
 * @return void* 空
 */
void *task_receive_data(void *arg)
{
    int sockfd = *(int *)arg;
    printf("\t[child thread] waiting data..\n");

    while (1)
    {
        // 设置互斥锁, 保证只有一个线程修改滑动窗口结构体
        while(1)
        {
            if(pthread_mutex_trylock(&lock))
            {
                break;
            }
        }

        // 接收窗口为空则跳过一轮接收
        if (receive_window->length == 0)
        {
            // 设置互斥锁, 保证只有一个线程修改滑动窗口结构体
            pthread_mutex_unlock(&lock);
            continue;
        }
        
        // 接收RDT数据包
        Queue_Element_Type e;
        e.length = recv(sockfd, e.packet, RDT_PACKET_LENGTH, 0);
        begin_flag = 1;

        // 解封装RDT数据包
        enum Enum_RDT_State receive_state;
        char receive_rdt_data[RDT_DATA_LENGTH];
        int receive_data_length = Unpack_RDT_Packet(e.packet, &receive_state, &e.sequence, receive_rdt_data, e.length);
        total_receive_byte += receive_data_length;
        
        // RDT数据包塞入队列
        struct Struct_Queue_Node *ptr_q = receive_window->front;
        while(ptr_q != NULL)
        {
            // 如果序号相等则填充对应数据到队列
            if(e.sequence == ptr_q->data.sequence)
            {
                ptr_q->data.length = e.length;
                memcpy(ptr_q->data.packet, e.packet, sizeof(char) * RDT_PACKET_LENGTH);
                ptr_q->data.state = RDT_Packet_Window_State_RECEIVED;
                break;
            }
            ptr_q = ptr_q->next;
        }

        // 如果接收到了结束包则标记位置
        if(receive_state == RDT_State_END)
        {
            finish_ptr = ptr_q;
        }

        // 设置互斥锁, 保证只有一个线程修改滑动窗口结构体
        pthread_mutex_unlock(&lock);
    }
}

/**
 * @brief 被task_send_acks()函数调用, 写入文件, 定位NACK位置, 滑动窗口
 * 
 * @param fp 被接收的文件
 * @return int 如果窗口内有需要ACK的包则返回1, 否则0
 */
int prepare_receive_window(FILE *fp)
{
    // 如果窗口不空, 则滑动窗口左侧更改队首, 接收的数据包出队, 写入文件
    while (receive_window->length != 0 && receive_window->front->data.state == RDT_Packet_Window_State_RECEIVED)
    {
        // 如果结束包位置已标记则激活结束标记
        if(finish_ptr == receive_window->front)
        {
            finish_flag = 1;
        }


        // 接收包按顺序出队
        Queue_Element_Type e;
        Queue_Depart(receive_window, &e);
        
        // 解封装RDT数据包
        enum Enum_RDT_State receive_state;
        char receive_rdt_data[RDT_DATA_LENGTH];
        int receive_data_length = Unpack_RDT_Packet(e.packet, &receive_state, &e.sequence, receive_rdt_data, e.length);

        // 数据写入文件
        fwrite(receive_rdt_data, sizeof(char), receive_data_length, fp);
        valid_receive_byte += receive_data_length;

        last_slide_sequence++;
    }

    // 填充窗口右侧数据到队尾, 直到队列满
    while(receive_window->length < RDT_WINDOWS_LENGTH)
    {
        // 只有序号和状态的空队员
        Queue_Element_Type e;
        e.length = 0;
        if(receive_window->length == 0)
        {
            // 窗口空则从最新滑动位置寻找序列号
            e.sequence = last_slide_sequence + 1;
        }
        else
        {
            // 窗口不空则直接从队尾寻找序列号
            e.sequence = receive_window->rear->data.sequence + 1;
        }
        e.state = RDT_Packet_Window_State_WAIT;

        // 数据包加入队列
        Queue_Enter(receive_window, e);
    }

    return receive_window->length;
}


/**
 * @brief 接收端接收函数
 *
 * @param receive_file_name 保存文件名
 * @param sockfd 接收数据的socket
 * @return int 0表示不出错, 否则出错
 */
int receive_file(char *receive_file_name, int sockfd)
{
    // 初始化互斥量
    pthread_mutex_init(&lock, NULL);

    // 创建子线程用于接收数据
    pthread_t thread_receive_data;
    if (pthread_create(&thread_receive_data, NULL, task_receive_data, (void *)&sockfd) != 0)
    {
        perror("pthread_create failed.\n");
        exit(1);
    }

    // 将收到的RDT数据包按顺序写到此文件中
    FILE *fp;
    if ((fp = fopen(receive_file_name, "wb")) == NULL)
    {
        printf("open file : %s failed.\n", receive_file_name);
        return 1;
    }

    // 接收RDT数据包, 直到所有数据全部接收完毕
    while (1)
    {
        // 休眠 us
        usleep(RDT_SEND_FREQUENCY);

        // 设置互斥锁, 保证只有一个线程修改滑动窗口结构体
        while(1)
        {
            if(pthread_mutex_trylock(&lock))
            {
                break;
            }
        }

        int has_packet_to_ack = prepare_receive_window(fp);

        // 还没开始接收到数据则不发送任何数据, 防止瞎发一通NACK闹心
        if(begin_flag == 0)
        {
            // 设置互斥锁, 保证只有一个线程修改滑动窗口结构体
            pthread_mutex_unlock(&lock);
            continue;
        }

        // 封装NACK包, 实现累计确认
        char send_rdt_packet[RDT_PACKET_LENGTH];
        enum Enum_RDT_State send_state = RDT_State_NACK;
        int send_sequence = last_slide_sequence + 1;
        int send_packet_length = Pack_RDT_Packet(send_rdt_packet, send_state, send_sequence, NULL, 0);

        // 调用不可靠数据传输发送新的RDT数据包
        RDT_Send(sockfd, send_rdt_packet, send_packet_length, 0);

        // 如果需要ACK则遍历队列进行发送
        if (has_packet_to_ack > 0)
        {
            // 遍历队列进行ACK
            for (struct Struct_Queue_Node *ptr_q = receive_window->front; ptr_q != NULL; ptr_q = ptr_q->next)
            {
                struct Struct_Window_Packet *ptr_packet = &ptr_q->data;

                // 判断哪些数据包需要ACK
                if(ptr_packet->state == RDT_Packet_Window_State_RECEIVED)
                {
                    // 发送ACK包

                    // 封装ACK包
                    char send_rdt_packet[RDT_PACKET_LENGTH];
                    enum Enum_RDT_State send_state = RDT_State_ACK;
                    int send_sequence = ptr_packet->sequence;
                    int send_packet_length = Pack_RDT_Packet(send_rdt_packet, send_state, send_sequence, NULL, 0);

                    // 调用不可靠数据传输发送新的RDT数据包
                    RDT_Send(sockfd, send_rdt_packet, send_packet_length, 0);
                }
            }
        }

        // 准备结束会话
        if(finish_flag == 1)
        {
            // 设置互斥锁, 保证只有一个线程修改滑动窗口结构体
            pthread_mutex_unlock(&lock);
            break;
        }

        // 设置互斥锁, 保证只有一个线程修改滑动窗口结构体
        pthread_mutex_unlock(&lock);
    }

    printf("\n\n");
    printf("Receive finished. Write to file %s.\n", receive_file_name);
    printf("Total recv %5d bytes\n", total_receive_byte);
    printf("Valid recv %5d bytes\n", valid_receive_byte);

    // 关闭文件并写入
    fflush(fp);
    fclose(fp);

    //销毁子线程
    if (pthread_cancel(thread_receive_data) != 0)
    {
        printf("[main thread] pthread_cancel failed.\n");
        exit(1);
    }

    //删除互斥量
    pthread_mutex_destroy(&lock);
    return 0;
}

int main(int argc, char **argv)
{
    if (argc != 2)
    {
        printf("wrong argument!\n");
        printf("usage: %s receive_file_name\n", argv[0]);
        exit(0);
    }

    Queue_Init(receive_window);
    int sockfd = Init_Socket_Receiver();
    if (receive_file(argv[1], sockfd) != 0)
    {
        printf("receive file %s failed.\n", argv[1]);
        close(sockfd);
        exit(1);
    }

    // Queue_Init(receive_window);
    // int sockfd = Init_Socket_Receiver();
    // char recv[5] = "recv";
    // if (receive_file(recv, sockfd) != 0)
    // {
    //     printf("receive file %s failed.\n", argv[1]);
    //     close(sockfd);
    //     exit(1);
    // }

    Queue_Destroy(receive_window);
    close(sockfd);
    return 0;
}
